﻿//========================================================= 
/**@file  CommandAgent.h
 * @brief  
 * 
 * @date 2024-10-22 15:42:49 
 * @author yingbaizhou@tencent.com 
 * @version 1.0
 */ 
//--------------------------------------------------------- 
#ifndef _LIBZHOUYB_COMMANDAGENT_H_
#define _LIBZHOUYB_COMMANDAGENT_H_
//--------------------------------------------------------- 
#define INCLUDE_SOCKET_SOURCE
#define INCLUDE_ASYNC_SOURCE

#include "CommandDriver.h"

#include "../../wrapper/rapidjson/rapidjson_extractor.h"
using namespace zhou_yb::wrapper::rapidjson_extractor;
//---------------------------------------------------------
#define LOG LogCat_Default(__FUNCTION__)
//--------------------------------------------------------- 
namespace zhou_yb {
namespace application {
namespace driver {
//--------------------------------------------------------- 
struct AgentTrans
{
    uint id;
    Timer::TickType timestamp;
};
//--------------------------------------------------------- 
class CommandAgent : public InterruptBehavior, public CommandCollection
{
public:
    //----------------------------------------------------- 
    enum CommandType
    {
        enum_CmdUnknown = 0,
        enum_ServerCall,
        enum_RemoteCall
    };
    enum MessageType
    {
        enum_MsgUnknown = 0,
        enum_MsgReq,
        enum_MsgRes,
        enum_MsgNotify
    };
    //----------------------------------------------------- 
    /**
     * @brief 打包指定的remote_call调用参数
     * @date 2019-03-14 16:02
     */
    static size_t PackCMD(const ByteArray& command, uint id, CommandType cmdType, MessageType msgType,
        uint from, uint to, const ByteArray& sJson, ByteBuilder& cmd)
    {
        size_t lastLen = cmd.GetLength();
        DevCommand::FromAscii("01 A1 00000000", cmd);

        size_t cmdlen = command.GetLength() + 1;
        size_t datalen = sJson.GetLength() + 1;
        size_t bodylen = 4 + cmdlen + 4 + (1 + 1 + 4 + 4) + 4 + datalen;
        // 4=命令头长度 4=ID 4=DATA长度
        size_t len = cmd.GetLength() + 4 + bodylen;
        ByteConvert::FromObject<size_t>(len, cmd, 4);

        // CMD
        ByteConvert::FromObject<size_t>(cmdlen, cmd, 4);
        cmd += command;
        cmd += static_cast<byte>(0x00);
        // ID
        ByteConvert::FromObject<size_t>(id, cmd, 4);
        // CommandType
        cmd += static_cast<byte>(cmdType);
        // MessageType
        cmd += static_cast<byte>(msgType);
        // From
        ByteConvert::FromObject<uint>(from, cmd, 4);
        // To
        ByteConvert::FromObject<uint>(to, cmd, 4);
        // DATA
        ByteConvert::FromObject<size_t>(datalen, cmd, 4);
        cmd += sJson;
        cmd += static_cast<byte>(0x00);
        return (cmd.GetLength() - lastLen);
    }
    static bool IsFormat(const ByteArray& data, byte& bCmd, size_t& len)
    {
        if (data.GetLength() < 10)
            return false;
        size_t bodylen = 0;
        ByteConvert::ToObject<size_t>(data.SubArray(6, 4), bodylen);
        if (bodylen > data.GetLength())
            return false;
        bCmd = data[1];
        len = bodylen;
        return true;
    }
    static bool UnpackCMD(const ByteArray& data, ByteArray& cmd, uint& id, CommandType& cmdType,
        MessageType& msgType, uint& from, uint& to, ByteArray& rJson)
    {
        byte bCmd = 0x00;
        size_t bodylen = 0;
        if (!IsFormat(data, bCmd, bodylen) || bCmd != 0xA1)
            return false;

        size_t cmdlen = 0;
        ByteArray body = data.SubArray(6 + 4);

        // Command
        uint offset = 0;
        uint sublen = 4;
        ByteConvert::ToObject<size_t>(body.SubArray(offset, sublen), cmdlen);
        offset += sublen;
        cmd = body.SubArray(offset, cmdlen);
        offset += cmdlen;

        // ID
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), id);
        offset += sublen;
        // CmdType
        sublen = 1;
        uint tmp = 0;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), tmp);
        cmdType = static_cast<CommandType>(tmp);
        offset += sublen;
        // MsgType
        sublen = 1;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), tmp);
        msgType = static_cast<MessageType>(tmp);
        offset += sublen;
        // From
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), from);
        offset += sublen;
        // To
        sublen = 4;
        ByteConvert::ToObject<uint>(body.SubArray(offset, sublen), to);
        offset += sublen;
        // BodyLen
        sublen = 4;
        ByteConvert::ToObject<size_t>(body.SubArray(offset, sublen), bodylen);
        offset += sublen;
        // Body
        rJson = body.SubArray(offset, bodylen);
        return true;
    }
    static bool RecvCMD(IInteractiveTrans& dev, ByteBuilder& data)
    {
        const size_t MIN_LEN = 10;
        if (!DevHelper::IsHelperSuccess(IInteractiveTransHelper::ReadAssert(dev, MIN_LEN, data)))
            return false;
        size_t len = 0;
        ByteConvert::ToObject(data.SubArray(6, 4), len);
        return DevHelper::IsHelperSuccess(IInteractiveTransHelper::ReadAssert(dev, len, data));
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    /// 心跳间隔
    uint HeartBeatIntervalMs;
    /// 事务超时时间
    uint TransTimeoutMs;
    /// 空闲等待间隔
    uint IdleIntervalMs;
    //----------------------------------------------------- 
protected:
    //----------------------------------------------------- 
    ByteBuilder _cachedBuff;
    AsyncSocketDevice _dev;
    string _ip;
    uint _port;
    uint _tid;
    uint _idx;
    uint _req_idx;
    RapidJsonArgParser _parser;

    std::list<AgentTrans> _trans_pool;
    uint _heart_beat_last_tid;
    uint _heart_beat_time_tick;
    bool _is_running;
    //----------------------------------------------------- 
    template<class T>
    static T config_option(IniGroup& grp, const char* key, const T& default_val = T())
    {
        IniGroup::const_iterator itr = grp.Get(key);
        if (itr == grp.end())
        {
            return default_val;
        }
        T val;
        if (!ArgConvert::FromString<T>(itr->Value, val))
            return default_val;
        return val;
    }
    //----------------------------------------------------- 
    bool _Wait(ByteBuilder& command, ByteBuilder& data, uint& id, uint& from, uint& to, CommandType& cmdType, MessageType& msgType)
    {
        byte bCmd = 0x00;
        size_t len = 0;
        // 没有历史数据,触发读
        if (!IsFormat(_cachedBuff, bCmd, len))
        {
            if (!RecvCMD(_dev, _cachedBuff))
                return false;
            if (!IsFormat(_cachedBuff, bCmd, len))
                return false;
        }
        if (bCmd == 0xA1)
        {
            ByteArray cmd;
            ByteArray sJson;
            UnpackCMD(_cachedBuff, cmd, id, cmdType, msgType, from, to, sJson);
            if (cmd[cmd.GetLength() - 1] == 0)
                cmd = cmd.SubArray(0, cmd.GetLength() - 1);
            if (sJson[sJson.GetLength() - 1] == 0)
                sJson = sJson.SubArray(0, sJson.GetLength() - 1);
            LOG.debug() << "recv " << cmd.GetString() << " id:" << id << ' ' << from << "->" << to << ' ' << sJson.GetString();
            command.Append(cmd);
            data.Append(sJson);
        }
        else
        {
            LOG.warn() << "recv error cmd : " << _hex_num(bCmd);
        }
        _cachedBuff.RemoveFront(len);
        return true;
    }
    bool _Send(const ByteArray& cmd, CommandType cmdType, MessageType msgType = enum_MsgReq, uint to = 0, IStringArguments* arg = NULL)
    {
        ByteBuilder send(64);
        ByteBuilder param(64);

        if (arg != NULL)
        {
            _parser.ToString(*arg, param);
        }
        if (msgType == enum_MsgNotify)
        {
            LOG.info() << "notify :" << cmd.GetString() << " \"" << param.GetString() << "\"";
        }
        LOG.debug() << "send " << cmd.GetString() << " id:" << (_tid + 1) << ' ' << _idx << "->" << to << ' ' << param.GetString();

        PackCMD(cmd, _tid + 1, cmdType, msgType, _idx, to, param, send);
        if (!_dev.Write(send))
            return false;
        AgentTrans t;
        _tid += 1;
        t.id = _tid;
        t.timestamp = Timer::TimeTick();

        _trans_pool.push_back(t);
        return true;
    }
    bool _SendNotify(const ByteArray& cmd, IStringArguments* arg = NULL)
    {
        return _Send(cmd, enum_RemoteCall, enum_MsgNotify, _req_idx, arg);
    }
    bool _HandleReq(const ByteArray& cmd, IStringArguments& arg, IStringArguments& rlt)
    {
        Ref<ComplexCommand> command = CommandCollection::LookUp(cmd);
        if (command.IsNull())
        {
            string msg = "<";
            msg += cmd.GetString();
            msg += "> is not exists";
            rlt.Put("Msg", msg);
            return false;
        }
        return command->OnCommand(arg, rlt);
    }
    //----------------------------------------------------- 
    virtual void _Update()
    {
       
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    /**
     * @brief 以守护进程的方式运行
     *
     * @param [in] url : 连接的服务器地址 127.0.0.1:9999
     * @param [in] cache : 配置文件及日志输出路径
     * @param [in] agent_ini : 配置文件名(不含扩展名)
     *
     * @date 2021-02-20 16:50
     */
    static bool domain(CommandAgent& agent, const char* url, const char* cache, const char* agent_ini)
    {
        IniFile ini;
        string config_path = _strput(cache);
        if (config_path.length() > 0)
        {
            config_path += PATH_SEPARATOR;
        }
        config_path += (_is_empty_or_null(agent_ini) ? "agent" : agent_ini);
        config_path += ".ini";

        string logdir = _strput(cache);
        if (logdir.length() > 0)
        {
            logdir += PATH_SEPARATOR;
        }
        logdir += "log";
        LogCatHelper::DefaultWithFolder(logdir.c_str(), "Agent");

        if (!ini.Open(config_path.c_str()))
        {
            LOG.warn() << "can't found config file : " << config_path;
        }
        IniGroup& grp = ini["agent"];
        string ip = "";
        ushort port = 0;
        ByteBuilder url_data(url);
        size_t index = StringConvert::IndexOf(url_data, ':');
        if (index != SIZE_EOF)
        {
            port = ArgConvert::FromString<ushort>(url_data.SubArray(index + 1).GetString());
            url_data[index] = 0;
            ip = url_data.GetString();
        }
        else
        {
            ip = config_option<string>(grp, "IP", "");
            port = config_option<ushort>(grp, "Port", 0);
        }
        uint timeoutMs = config_option<uint>(grp, "TimeoutMs", 10000);
        string sLogLv = config_option<string>(grp, "LogLv", "INFO");
        LogCatHelper::Default().SetPriority("Console", LogCat::ConvertFrom(sLogLv.c_str()));

        if (ip.length() < 1 || port == 0)
        {
            LOG.error() << "IP or Port is error (" << ip << ':' << port << ")";
            return false;
        }

        bool is_auto_init = config_option<bool>(grp, "IsAutoInit", false);
        if (is_auto_init && !agent.Init(ini))
        {
            LOG.error() << "Agent init failed.";
            return false;
        }

        string name = config_option<string>(grp, "Name", "none");
        uint max_retry_count = config_option<uint>(grp, "RetryCount", 3);

        agent.TransTimeoutMs = config_option<uint>(grp, "TransTimeoutMs", 3000);
        agent.IdleIntervalMs = config_option<uint>(grp, "IdleIntervalMs", 200);
        agent.HeartBeatIntervalMs = config_option<uint>(grp, "HeartBeatIntervalMs", 3000);

        uint retry_count = 0;
        while (retry_count < max_retry_count)
        {
            retry_count += 1;
            if (!agent.Connect(ip.c_str(), port, timeoutMs))
            {
                LOG.error() << "connect to agent server timeout. retry count : " << retry_count;
                if (retry_count < max_retry_count)
                {
                    Timer::Wait(agent.TransTimeoutMs);
                }
                continue;
            }
            else
            {
                retry_count = 0;
            }
            if (agent.Run(name.c_str()))
                break;
            LOG.error() << "agent is error, reconnect: " << retry_count;
        }
        LOG.warn() << "exit to domain.";
        return true;
    }
    //----------------------------------------------------- 
public:
    //----------------------------------------------------- 
    CommandAgent()
    {
        _ip = "";
        _port = 0;
        _tid = 0;
        _idx = 0;

        _heart_beat_time_tick = 0;
        _heart_beat_last_tid = 0;
        _is_running = false;

        HeartBeatIntervalMs = 3000;
        TransTimeoutMs = 3000;
        IdleIntervalMs = 100;
    }
    //----------------------------------------------------- 
    bool Connect(const char* ip, ushort port, uint timeoutMs)
    {
        LOG.info() << _expression(ip) << ' ' << _expression(port) << ' ' << _expression(timeoutMs);
        const uint connect_timeoutMs = 1000;
        _dev.SetWaitTimeout(connect_timeoutMs);
        _dev.Interrupter = Interrupter;

        // Connect to Server
        Timer timer;
        while (timer.Elapsed() < timeoutMs)
        {
            if (DevHelper::IsHelperSuccess(SocketDeviceHelper::Connect(_dev, ip, port)))
            {
                LOG.info() << "Connect to Server Success.";
                return true;
            }
            else
            {
                LOG.warn() << "Connect agent server failed.";
            }
            Timer t;
            while (t.Elapsed() < connect_timeoutMs)
            {
                /* 中断支持 */
                if (InterruptBehavior::Implement(*this))
                {
                    LOG.error() << "Connecting is Canceled.";
                    return false;
                }
                Timer::Wait(100);
            }
        }
        LOG.error() << "Connect to Server Timeout.";
        return false;
    }
    /**
     * @brief 初始化配置
     * @date 2021-02-26 20:30
     */
    bool Init(IniFile& cfg)
    {
        StringArguments arg;
        StringArguments rlt;
        ByteBuilder buff(128);
        IniGroup& grp = cfg["auto_init"];

        for (IniGroup::iterator itr = grp.begin(); itr != grp.end(); ++itr)
        {
            if(itr->Key.length() < 1)
                continue;

            LOG.info() << "auto init : " << itr->Key;
            ByteArray cmd(itr->Key.c_str(), itr->Key.length());

            arg.Clear();
            rlt.Clear();
            buff.Clear();

            if (itr->Value.length() > 0)
            {
                IniGroup& param_grp = cfg[itr->Value];
                for (IniGroup::iterator param_itr = param_grp.begin(); param_itr != param_grp.end(); ++param_itr)
                {
                    arg.Put(param_itr->Key, param_itr->Value);
                }
                _parser.ToString(arg, buff);
                LOG.info() << buff.GetString();
            }

            bool bOK = _HandleReq(cmd, arg, rlt);

            buff.Clear();
            _parser.ToString(rlt, buff);
            LOG.info() << "IsOK:" << bOK << ' ' << buff.GetString();
        }

        return true;
    }
    /**
     * @brief 接收服务端数据,返回是否正常退出
     * @date 2021-02-05 17:16
     */
    bool Run(const char* name = NULL)
    {
        ByteBuilder cmd(16);
        ByteBuilder data(128);

        uint tid = 0;
        uint from = 0;
        uint to = 0;
        CommandType cmdType = enum_CmdUnknown;
        MessageType msgType = enum_MsgUnknown;

        StringArguments arg;
        StringArguments rlt;
        bool bOK = false;
        if (!_is_empty_or_null(name))
        {
            arg.Put("Name", name);
        }
        _Send("agent_svr_heart_beat", enum_ServerCall, enum_MsgReq, 0, &arg);
        _heart_beat_last_tid = _tid;
        _heart_beat_time_tick = Timer::TimeTick() + HeartBeatIntervalMs;
        _dev.SetWaitTimeout(IdleIntervalMs);
        _is_running = true;
        while (_is_running)
        {
            Timer::TickType time_tick = Timer::TimeTick();
            // 中断支持
            if (InterruptBehavior::Implement(*this))
                break;

            _Update();

            for (std::list<AgentTrans>::iterator itr = _trans_pool.begin(); itr != _trans_pool.end();)
            {
                if (itr->timestamp >= (time_tick + TransTimeoutMs))
                {
                    LOG.error() << "trans is timeout : " << itr->id;
                    if (itr->id == _heart_beat_last_tid)
                    {
                        LOG.error() << "agent server maybe is busy.";
                        return true;
                    }

                    itr = _trans_pool.erase(itr);
                }
                else
                {
                    ++itr;
                }
            }

            cmd.Clear();
            data.Clear();
            if (!_Wait(cmd, data, tid, from, to, cmdType, msgType))
            {
                // 当前有通信数据时不触发心跳
                if (time_tick > _heart_beat_time_tick)
                {
                    if (!_Send("agent_svr_heart_beat", enum_ServerCall))
                    {
                        LOG.error() << "send message to agent server failed, server maybe is closed.";
                        return false;
                    }
                    _heart_beat_last_tid = _tid;
                    _heart_beat_time_tick = time_tick + HeartBeatIntervalMs;
                }
                else
                {
                    Timer::Wait(IdleIntervalMs);
                }
                continue;
            }
            // Agent客户端只处理 RemoteCall
            if (cmdType != enum_RemoteCall && msgType == enum_MsgReq)
            {
                LOG.error() << "recv error cmd_type : " << static_cast<int>(cmdType);
                continue;
            }
            switch (msgType)
            {
            case enum_MsgReq:
                _req_idx = from;

                arg.Clear();
                rlt.Clear();

                LOG.info() << "on call : " << cmd.GetString() << " \"" << data.GetString() << "\"";
                if (data.GetLength() > 0)
                {
                    _parser.Parse(data, arg);
                }
                bOK = _HandleReq(cmd, arg, rlt);
                StringArguments::Set(rlt, "IsOK", ArgConvert::ToString(bOK));
                _SendNotify(cmd, &rlt);
                break;
            case enum_MsgRes:
                for (std::list<AgentTrans>::iterator itr = _trans_pool.begin(); itr != _trans_pool.end(); ++itr)
                {
                    if (itr->id == tid)
                    {
                        _trans_pool.erase(itr);
                        break;
                    }
                }
                if (StringConvert::Compare(cmd, "agent_svr_heart_beat"))
                {
                    _idx = to;
                }
                break;
            default:
                LOG.error() << "recv error msg_type : " << static_cast<int>(msgType);
                break;
            }
        }
        return true;
    }
    //----------------------------------------------------- 
};
//--------------------------------------------------------- 
} // namespace driver
} // namespace application 
} // namespace zhou_yb
//--------------------------------------------------------- 
#endif // _LIBZHOUYB_COMMANDAGENT_H_
//=========================================================